package com.java1234.rocketmq;

import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;

/**
 * @author zhaolin
 * @version 1.0
 * @description: TODO 消息生产者
 * @date 2025/2/6 22:17
 */
@Component("producerService")
public class ProducerService {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;


    /**
     * 发送简单消息
     */
    public void sendMessage() {
        for (int i = 0; i < 10; i++) {
            rocketMQTemplate.convertAndSend("java1234-rocketmq", "rocketmq大爷，你 好！" + i);
        }
    }

    /**
     * 发送同步消息
     */
    public void sendSyncMessage() {
        for (int i = 0; i < 10; i++) {
            SendResult sendResult = rocketMQTemplate.syncSend("java1234-rocketmq", "rocketmq同步消息！" + i);
            System.out.println(sendResult);
        }
    }


    /**
     * 发送异步消息
     */
    public void sendAsyncMessage() {
        for (int i = 0; i < 10; i++) {
            rocketMQTemplate.asyncSend("java1234-rocketmq", "rocketmg异步消息！" + i,
                    new SendCallback() {
                        @Override
                        public void onSuccess(SendResult sendResult) {
                            System.out.println("发送成功！");
                        }

                        @Override
                        public void onException(Throwable throwable) {
                            System.out.println("发送失败！");
                        }
                    });
        }
    }

    /**
     * 发送单向消息
     * 例如日志发送
     */
    public void sendOneWayMessage() {
        for (int i = 0; i < 10; i++) {
            rocketMQTemplate.sendOneWay("java1234-rocketmq", "rocketmg单向消息！" + i);
        }
    }


    /**
     * 发送同步顺序消息
     */
    public void sendOrderlyMessage() {
// hashKey用来计算决定消息发送到哪个消息队列 一般是订单ID，产品ID等
        rocketMQTemplate.syncSendOrderly("java1234-rocketmq-orderly",
                "98456231,创 建",
                "98456231");
        rocketMQTemplate.syncSendOrderly("java1234-rocketmq-orderly", "98456231,支 付", "98456231");
        rocketMQTemplate.syncSendOrderly("java1234-rocketmq-orderly", "98456231,完 成", "98456231");
        rocketMQTemplate.syncSendOrderly("java1234-rocketmq-orderly", "98456232,创 建", "98456232");
        rocketMQTemplate.syncSendOrderly("java1234-rocketmq-orderly", "98456232,支 付", "98456232");
        rocketMQTemplate.syncSendOrderly("java1234-rocketmq-orderly", "98456232,完 成", "98456232");
    }

    /**
     * 发送延迟消息
     */
    public void sendDelayMessage() {
        rocketMQTemplate.syncSend("java1234-rocketmq-orderly", MessageBuilder.withPayload(" rocketmq延迟1秒消息").build(),3000,1);
        rocketMQTemplate.syncSend("java1234-rocketmq-orderly",MessageBuilder.withPayload(" rocketmq延迟5秒消息").build(),3000,2);
        rocketMQTemplate.syncSend("java1234-rocketmq-orderly",MessageBuilder.withPayload(" rocketmq延迟10秒消息").build(),3000,3);
    }


    /**
     * 发送事务消息
     */
    public void sendTransactionMessage(){
// 构造消息
        Message msg = MessageBuilder.withPayload("rocketmq事务消息-01").build();
        rocketMQTemplate.sendMessageInTransaction("java1234-rocketmq-orderly",msg,null);
    }

    @RocketMQTransactionListener
    class TransactionListenerImpl implements RocketMQLocalTransactionListener {
        /**
         * 返回事务执行状
         * 态，有 bollback , commit or unknown 三种，分别是回滚事务，提交事务和未知；
         * @param msg
         * @param arg
         * @return
         */
        @Override
        public RocketMQLocalTransactionState executeLocalTransaction(Message msg,
                                                                     Object arg) {
// ... local transaction process, return bollback, commit or unknown
            System.out.println("executeLocalTransaction");
            return RocketMQLocalTransactionState.UNKNOWN;
        }

        /**
         * 是当MQ Server未得到MQ发送方应答，或者超时的情况，或者应答是
         * unknown的情况，调用此方法进行检查确认，返回值和上面的方法一样；
         * @param msg
         * @return
         */
        @Override
        public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
// ... check transaction status and return bollback, commit or unknown
            System.out.println("checkLocalTransaction");
            return RocketMQLocalTransactionState.COMMIT;
        }
    }

    /**
     * 发送带Tag消息，测试根据Tag过滤消息
     */
    public void sendMessageWithTag() {
        // 构造消息1
        Message msg1 = MessageBuilder.withPayload("rocketmq过滤消息测试- TAG01").build();
        // 构造消息2
        Message msg2 = MessageBuilder.withPayload("rocketmq过滤消息测试- TAG02").build();
        // 构造消息3
        Message msg3 = MessageBuilder.withPayload("rocketmq过滤消息测试- TAG03").build();

        rocketMQTemplate.convertAndSend("java1234-filter-rocketmq" + ":" + "TAG1", msg1);
        rocketMQTemplate.convertAndSend("java1234-filter-rocketmq" + ":" + "TAG2", msg2);
        rocketMQTemplate.convertAndSend("java1234-filter-rocketmq" + ":" + "TAG3", msg3);
    }


    /**
     * 发送SQL表达式头信息消息，测试根据SQL表达式过滤消息
     * 分别带上不同的header头信息
     */
    public void sendMessageWithSQL(){
// 构造消息1
        Message msg1 = MessageBuilder.withPayload("rocketmq过滤消息测试01").build();
        Map<String, Object> headers = new HashMap<>() ;
        headers.put("type", "pay") ;
        headers.put("a", 10) ;
        rocketMQTemplate.convertAndSend("java1234-filter-rocketmq", msg1, headers);
// 构造消息2
        Message msg2 = MessageBuilder.withPayload("rocketmq过滤消息测试02").build();
        Map<String, Object> headers2 = new HashMap<>() ;
        headers2.put("type", "store") ;
        headers2.put("a", 4) ;
        rocketMQTemplate.convertAndSend("java1234-filter-rocketmq", msg2, headers2);
// 构造消息3
        Message msg3 = MessageBuilder.withPayload("rocketmq过滤消息测试03").build();
        Map<String, Object> headers3 = new HashMap<>() ;
        headers3.put("type", "user") ;
        headers3.put("a", 7) ;
        rocketMQTemplate.convertAndSend("java1234-filter-rocketmq", msg3, headers3);
    }

}
